package com.inventive.jmscache.server;

import javax.jms.*;
import java.lang.IllegalStateException;

/**
 * @author Grant Little <grant.little@coreservices.com.au>
 */
public class JmsCacheMessageConsumer implements MessageConsumer {

    protected JmsCacheDestination destination;

    protected MessageListener messageListener;
    protected String messageSelector;
    protected boolean closed = false;
    private AcknowledgeMode acknowledgeMode;
    private JmsCacheSession session;
    //If destination of consumer is a topic then inhibits delivery of messages from its own connection
    //http://download.oracle.com/javaee/1.4/api/javax/jms/Session.html#createConsumer%28javax.jms.Destination,%20java.lang.String,%20boolean%29
    boolean noLocal=false;

    public JmsCacheMessageConsumer(JmsCacheSession session, JmsCacheDestination destination) {
        this.session = session;
        this.destination = destination;
        acknowledgeMode = AcknowledgeMode.SESSION_TRANSACTED;
    }

    public JmsCacheMessageConsumer(JmsCacheSession session, JmsCacheDestination destination, AcknowledgeMode acknowledgeMode) {
        this.session = session;
        this.destination = destination;
        this.acknowledgeMode = acknowledgeMode;
    }

    public JmsCacheMessageConsumer(JmsCacheSession session, JmsCacheDestination destination, String messageSelector, boolean noLocal) {
        this.session = session;
        this.destination = destination;
        this.messageSelector = messageSelector;
        this.noLocal = noLocal;
        acknowledgeMode = AcknowledgeMode.SESSION_TRANSACTED;
    }

    public JmsCacheMessageConsumer(JmsCacheSession session, JmsCacheDestination destination, String messageSelector, AcknowledgeMode acknowledgeMode, boolean noLocal) {
        this.session = session;
        this.destination = destination;
        this.messageSelector = messageSelector;
        this.acknowledgeMode = acknowledgeMode;
        this.noLocal = noLocal;
    }


    public String getMessageSelector() throws JMSException {
        return messageSelector;
    }

    public MessageListener getMessageListener() throws JMSException {
        return messageListener;
    }

    public void setMessageListener(MessageListener messageListener) throws JMSException {
        this.messageListener = messageListener;
        if (null != messageListener) {
            while (!closed) {
                Message message = receive(5000);
                if (null != message) {
                    messageListener.onMessage(message);
                }
            }
        }
    }

    public Message receive() throws JMSException {

        if (closed) {
            throw new IllegalStateException("MessageConsumer is already closed");
        }
        if (acknowledgeMode == AcknowledgeMode.SESSION_TRANSACTED) {
            session.startTransaction();
        }

        try {
            Message message = doReceive(null);

            if (acknowledgeMode == AcknowledgeMode.SESSION_TRANSACTED) {
                session.commit();
            }
            return message;
        } catch (Exception e) {
            session.rollback();
            throw new JMSException(e.getMessage());
        }
    }


    protected Message doReceive(Long timeoutMillis) throws JMSException {
        if (null == timeoutMillis) {
            return destination.receive();
        } else {
            return destination.receive(timeoutMillis);
        }
    }

    public Message receive(long timeoutMillis) throws JMSException {
        if (closed) {
            throw new IllegalStateException("MessageConsumer is already closed");
        }
        if (acknowledgeMode == AcknowledgeMode.SESSION_TRANSACTED) {
            session.startTransaction();
        }

        try {
            Message message = doReceive(timeoutMillis);

            if (acknowledgeMode == AcknowledgeMode.SESSION_TRANSACTED) {
                session.commit();
            }
            return message;
        } catch (Exception e) {
            session.rollback();
            throw new JMSException(e.getMessage());
        }
    }

    public Message receiveNoWait() throws JMSException {
        return receive(0L);
    }

    public void close() throws JMSException {
        this.closed = true;
    }

}
